package regular;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.Queue;

import org.apache.hadoop.filecache.DistributedCache;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.BooleanWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Reducer;

public class EvalSecondReducer extends
		Reducer<IntWritable, NodeWritable, NullWritable, BooleanWritable> {

	public void reduce(IntWritable key, Iterable<NodeWritable> values,
			Context context) throws IOException, InterruptedException {
		boolean answer = false;

		Iterator<NodeWritable> itr = values.iterator();
		NodeWritable data = new NodeWritable();
		ArrayList<Node> listNode = new ArrayList<Node>();
		while (itr.hasNext()) {
			data = itr.next();
			listNode.addAll(data.listNode);
		}

		// get query
		Path[] fileCaches = DistributedCache.getLocalCacheFiles(context
				.getConfiguration());
		GraphQuery gq = Ultility.getAutomaton(fileCaches[0]);

		answer = evalDG(listNode, gq);

		context.write(NullWritable.get(), new BooleanWritable(answer));
	}

	// A BFS based function to check whether s can reach t
	public static boolean isReachable(ArrayList<Node> Vd, Node s, Node t) {
		if (s.equals(t))
			return true;

		// Create a queue for BFS
		Queue<Node> queue = new LinkedList<Node>();
		s.visit = true;
		queue.add(s);

		while (!queue.isEmpty()) {
			Node v = queue.poll();
			
			for (int e : v.neighbors) {
				// If this neighbor is the target vertex, then return true
				if (t.equals(new Node(e)))
					return true;
				// Else, continue to do BFS
				int id = Vd.indexOf(new Node(e));
				
				if (id >= 0) {
					if (Vd.get(id).visit == false) {
						queue.add(Vd.get(id));
						Vd.get(id).visit = true;						
					}
				}
			}
		}
		return false;
	}

	public boolean evalDG(ArrayList<Node> Vd, GraphQuery gq) {
		boolean answer = false;

		Node s = new Node(Integer.parseInt(gq.us.stateName));
		Node t = new Node(Integer.parseInt(gq.ut.stateName));

		int ids = Vd.indexOf(s);
		if(ids<0)
			return false;
		int idt = Vd.indexOf(t);
		if(idt<0){
			Vd.add(t);
			idt = Vd.size()-1;
		}			
		
		// BFS to get final answer
		answer = isReachable(Vd, Vd.get(ids), Vd.get(idt));
		return answer;
	}
}
